package com.shujia.flink.core

import org.apache.flink.api.common.functions.{ReduceFunction, RichMapFunction, RuntimeContext}
import org.apache.flink.api.common.state.{ReducingState, ReducingStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object Demo5ReducingStateWC {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val ds: DataStream[String] = env.socketTextStream("master", 7777)

    val kvDS: KeyedStream[(String, Long), String] = ds.flatMap(_.split(","))
      .map((_, 1L))
      .keyBy(_._1)

    val countDS: DataStream[(String, Long)] = kvDS.map(new MapFuncationState)

    countDS.print()


    env.execute()

  }

}

class MapFuncationState extends RichMapFunction[(String, Long), (String, Long)] {

  var countState: ReducingState[Long] = _

  override def open(parameters: Configuration): Unit = {
    val context: RuntimeContext = getRuntimeContext


    //创建一个聚合状态
    val reduceState: ReducingStateDescriptor[Long] = new ReducingStateDescriptor[Long]("countReduce", new ReduceFunction[Long] {
      override def reduce(value1: Long, value2: Long): Long = {
        value1 + value2
      }
    }, createTypeInformation[Long])

    //将状态加入到flink环境中  并返回状态对象
    countState = context.getReducingState(reduceState)

  }

  override def map(value: (String, Long)): (String, Long) = {

    //往状态上累加一个值
    countState.add(value._2)

    //获取状态中的值
    val count: Long = countState.get()

    //返回结果
    (value._1, count)
  }
}
